/* Author: cong
 * Date: Feb 15
 * File: client.c
 */  
/* the purpose of this file is to provide 
 * a class that reads vfile and efile and
 * produce an rfile that can be used as a
 * partition strategy
 */
/* here goes the definitions of vfile, efile and rfile
 * that is, the input of this module
 * suppose we have an undirected graph of vn vertices
 * and en edges.
 * 
 * vfile has vn lines with each line `vid a0 a1 ...'(no vsize && at least one a0)
 * efile has en lines with each line `src dst a0 a1 ...'(at least one a0)
 * rfile has vn lines with each line `vid fid'
 * 
 * note that the index of vertices starts from 1 (NOT 0)
 * the basic idea of this module is coding a series of
 * functions that converts from one representation to
 * another. Basically they are:
 * 
 * File: FILE *vfile, *efile, *rfile 
 * Vector: igraph_vector_ptr_t *edges, *vertices 
 * Igraph: igraph_t *igraph_graph
 * Adjlist: idx_t *metis_xadj, *metis_adjncy
 * 
 * for these functions:
 * 
 * precondition: the From params is garuanteed (inited, opened) by caller (but not necessarily 
 * correct in format, detecting errors is the responsibility of the functions)
 * 
 * postcondition: the correctness of To params is garuanteed by the functions 
 * if the From is correct then the To must be inited and correctly calculated
 */ 

#include "includes.h"
#include "structs.h"

/* errno of client.c */
int clerrno;
static void *xmalloc(int size)
{
        if(size <= 0)
                gk_errexit(SIGERR, "xmalloc: EIVALD\n");
        void *ptr = malloc(size);
        if(NULL == ptr)
                gk_errexit(SIGERR, "xmalloc: ENOMEM\n");
        return ptr;
}

/* Parse File vfile and store it in Vector vertices 
 * Input:
 * vfile: described as above
 * ncon: number of attributes attached to a single vertex
 * Ouput:
 * vn: pointer to number of vertices 
 * vertices: pointer to a  igraph_vector_ptr_t.
 * each entry points to a vertex_t structure
 * Time: O(vn)
 */

void print_vertex(FILE *s, struct vertex_t* v)
{
        fprintf(s,"vertex(vid=%d,ncon=%d,vsize=%d,attr=%p,fid=%d,locid=%d)fids=",
                        v->vid + 1,
                        v->ncon,
                        v->vsize,
                        v->attr,
                        v->fid,
                        v->locid);
        igraph_vector_int_fprint(v->fids, s);

}

void vertex_free(void *ptr)
{
        struct vertex_t *pv=ptr;
        igraph_vector_int_destroy(pv->fids);
        if(pv->ncon)
                free(pv->attr);
        free(pv);
}

int vfile2vertices(FILE *vfile, int ncon, igraph_vector_ptr_t *vertices)
{
        int vn = 0;
        int vid, iattr;
        char *buffer = NULL; 
        ssize_t read;
        size_t len;
        struct vertex_t *pvertex;

        igraph_vector_ptr_init(vertices, 0);
        igraph_vector_ptr_set_item_destructor(vertices,free);
        while(-1 != (read = getline(&buffer, &len, vfile))) 
        {
                FILE *sstream = fmemopen(buffer, len, "r");
                if(1 != fscanf(sstream, "%d", &vid) ) {
                        continue;
                }
                vn++;
                pvertex = (struct vertex_t*) xmalloc(sizeof(struct vertex_t));
                pvertex->vid = vid-1;
                pvertex->attr = (int*) calloc(sizeof(int), ncon);
                pvertex->ncon = ncon;
                pvertex->vsize=0;
                pvertex->fids=xmalloc(sizeof(igraph_vector_int_t));
                igraph_vector_int_init(pvertex->fids,0);
                pvertex->locid=0;
                pvertex->fid=0;
                for(int i=0;i<ncon;++i) 
                {
                        fscanf(sstream, "%d", &iattr);
                        pvertex->attr[i] = iattr;
                }
                igraph_vector_ptr_push_back(vertices, pvertex);
                fclose(sstream);
        }
        free(buffer);
        igraph_vector_ptr_set_item_destructor(vertices, vertex_free);

#ifndef NDEBUG
        for(int i=0;i<vn;i++) {
                struct vertex_t *pv = VECTOR(*vertices)[i];
                print_vertex(stderr, pv);
        }
#endif
        return vn;
}


/* for testing purpose wrapper */
int call_vfile2vertices(char *filename, int ncon, igraph_vector_ptr_t *vertices)
{
        zlog_category_t *mylog = zlog_get_category("mylog");
        FILE *vfile = fopen(filename, "r");
        if(!vfile) {
                zlog_error(mylog,  "vfile is null");
                exit(1);
        }
        if(!vertices) {
                zlog_error(mylog, "vertices is null");
                exit(1);
        }
        if( ncon < 1) {
                zlog_error(mylog, "ncon < 1");
                exit(1);
        }
        int vn = vfile2vertices(vfile, ncon , vertices);
        if (-1 == vn) {
                zlog_error(mylog, "vfile2vertices failed");
                exit(1);
        }

        zlog_info(mylog, "vfile2vertices succeeded , %d\n",vn);
        return vn;

}


void print_edge(FILE *s, struct edge_t *pedge)
{

#ifndef NDEBUG
        
        zlog_category_t *mylog = zlog_get_category("mylog");
        if(pedge->eid < 0) {
                zlog_error(mylog,"invalid eid %d", pedge->eid);
                abort();
        }
#endif
        fprintf(s, "edge(eid=%d,src=%d,dst=%d,attr=%d)\n",
                        pedge->eid + 1,
                        pedge->src + 1,
                        pedge->dst + 1,
                        pedge->attr);
}

/* Parse an efile and store these edges into edges
 * Input: efile: opened File described as above
 * Output: edges: Vector of edge_t
 * Time: O(en)
 */ 

int efile2edges(FILE *efile, igraph_vector_ptr_t *edges)
{
        int src, dst, attr;
        ssize_t read;
        size_t len;
        struct edge_t *pedge = NULL;
        int en=0;
        char *buffer = NULL;

        igraph_vector_ptr_init(edges, 0);
        igraph_vector_ptr_set_item_destructor(edges,free);
        while(-1 != (read = getline(&buffer, &len, efile))) 
        {
                if(3 != sscanf(buffer , "%d%d%d", &src, &dst, &attr)) 
                        continue;
                pedge = (struct edge_t*) xmalloc(sizeof(struct edge_t));
                pedge->src = src -1 ;
                pedge->dst = dst -1 ;
                pedge->attr = attr;
                assert(en >= 0);
                pedge->eid = en;
                en ++;
                igraph_vector_ptr_push_back(edges, pedge);
        }
        free(buffer);
#ifndef NDEBUG
        for(int i=0;i<en;i++) {
                struct edge_t *pedge = igraph_vector_ptr_e(edges,i);
                print_edge(stderr, pedge);
        }
#endif
        return en;
}

int call_efile2edges(const char *filename, igraph_vector_ptr_t *edges)
{
        int en;
        FILE *efile;
        zlog_category_t *mylog = zlog_get_category("mylog");
        efile = fopen(filename, "r");

        if( NULL == efile ) { 
                zlog_error(mylog, "failed to open efile");
                exit(1);
        } 
        en = efile2edges(efile, edges);
        if(en == -1){
                zlog_error(mylog, "efile2edges failed");
                exit(1);
        }
        zlog_info(mylog, "efile2edges succeeded, en=%d", en);
        return en;
}

/* Convert Vectors of edges and vertices to
 * igraph_adjlist_t(an array of adjlists)
 * note that the adjlist of each vertex is 
 * in fact of igraph_vector_int_t so only 
 * vertex id is retained in this structure
 * 
 * Input: as described in vector2adjlist
 * Ouput: iadjlist 
 * Time: O(en + vn)
 */ 


int vector2igraph_adjlist(int vn, int en, 
                igraph_vector_ptr_t *vertices,
                igraph_vector_ptr_t *edges,
                igraph_adjlist_t *iadjlist)


{
        igraph_adjlist_init_empty(iadjlist, vn);
        struct edge_t *pedge;
        int dst, src ;
        igraph_vector_int_t *adjlist;

        for(int i=0;i<en;++i) { 
                /* pedge = VECTOR(*edges)[i]; */
                pedge=igraph_vector_ptr_e(edges,i);
                src = pedge->src;
                dst= pedge->dst;
                adjlist = igraph_adjlist_get(iadjlist, src);
                igraph_vector_int_push_back(adjlist, dst);
                adjlist = igraph_adjlist_get(iadjlist, dst );
                igraph_vector_int_push_back(adjlist, src);

        }
#ifndef NDEBUG
        igraph_adjlist_fprint(iadjlist,stderr);
#endif
        return 0;
}


void my_adjlist_destory(my_adjlist_t *al)
{
        int vn=al->vn;
        
        for(int i=0;i<vn;++i) {
                igraph_vector_ptr_free_all(&(al->al[i]));
        }
        free(al->al);
}

int my_adjlist_init_empty(my_adjlist_t *al, int vn)
{
        zlog_category_t *mylog = zlog_get_category("mylog");
        int rc;

        if(vn <= 0) {
                zlog_error(mylog, "vn <= 0");
                return -1;
        }
        if(NULL == al) {
                zlog_error(mylog, "al is NULL");
                return -1;
        }
        al->al = xmalloc(sizeof(igraph_vector_ptr_t) * vn);
        for (int i=0;i<vn;++i) {
                rc = igraph_vector_ptr_init(&(al->al[i]), 0);
                igraph_vector_ptr_set_item_destructor(&(al->al[i]), free);
        }
        al->vn = vn;
        al->en = 0;
        return 0;
}

/* get the adjlist for vertex i
 * Input: al -> a bundle of adjlists 
 *      i -> vertex i
 * Ouput: adjlist of i
 * Error: NULL 
 */
igraph_vector_ptr_t *my_adjlist_get(my_adjlist_t *al, int i)
{
        zlog_category_t *mylog = zlog_get_category("mylog");
        if(i<0 || i>=al->vn) {
                zlog_error(mylog, "invalid i %d", i);
                return NULL;
        }
        return &(al->al[i]);
}

/* print textual info of a my_adjlist_t */
void my_adjlist_print(FILE *s, my_adjlist_t *al)
{
        int vn = al ->vn;
        fprintf(s, "my_adjlist_t(vn=%d)\n", al->vn);
        for(int i=0;i<vn;++i) {
                igraph_vector_ptr_t *pa = &(al->al[i]);
                int size = igraph_vector_ptr_size(pa);
                fprintf(s, "\tvid=%d\n", i+1);
                for(int j=0;j<size;++j) {
                        struct edge_t *pe = VECTOR(*pa)[j];
                        /* fprintf(s,"%d ", pe->dst+1); */
                        fprintf(s, "\t\t");
                        print_edge(s, pe);
                }
                fprintf(s, "\n");
        }
}

/* the same function that converts Vectors to 
 * Adjlist, but using the my_adjlist_t where
 * each element of an adjlist points to a
 * edge_t structure 
 */

struct edge_t *swap_and_store(struct edge_t *e)
{
        struct edge_t *pedge = xmalloc(sizeof(struct edge_t));
        memcpy(pedge, e, sizeof(struct edge_t));
        pedge->dst = e->src;
        pedge->src = e->dst;
        return pedge;
}

int vector2my_adjlist(
                igraph_vector_ptr_t *vertices,
                igraph_vector_ptr_t *edges,
                my_adjlist_t *iadjlist)


{
        int vn = igraph_vector_ptr_size(vertices);
        int en = igraph_vector_ptr_size(edges);

        int err = my_adjlist_init_empty(iadjlist, vn);
        if(err)
                return -1;
        struct edge_t *pedge;
        int dst, src ;
        igraph_vector_ptr_t *adjlist;

        for(int i=0;i<en;++i) { 
                /* pedge = VECTOR(*edges)[i]; */
                pedge=igraph_vector_ptr_e(edges,i);
                src = pedge->src;
                dst= pedge->dst;
                adjlist = my_adjlist_get(iadjlist, src);
                igraph_vector_ptr_push_back(adjlist, pedge);
                adjlist = my_adjlist_get(iadjlist, dst );
                /* make a new edges for this is an undirected graph */ 
                pedge = swap_and_store(pedge);
                igraph_vector_ptr_push_back(adjlist, pedge);

        }
#ifndef NDEBUG
        my_adjlist_print(stderr,iadjlist);
#endif
        return 0;
}

/* Convert my_adjlist_t to adjlist in METIS format 
 * Input: al: undirected adjlist whose element is struct edge_t
 * Ouput: xadj, adjncy, vwgt, adjwgt: 
 *      igraph_vector_int_t 
 */

int my_adjlist2adjlist(
                igraph_vector_ptr_t *vertices,
                my_adjlist_t *al,
                igraph_vector_int_t *xadj,
                igraph_vector_int_t *adjncy,
                igraph_vector_int_t *vwgt,
                igraph_vector_int_t *adjwgt)
{
        int vn = al->vn;                
        igraph_vector_int_init(xadj, 0);
        igraph_vector_int_init(adjncy, 0);
        igraph_vector_int_init(vwgt, 0);
        igraph_vector_int_init(adjwgt, 0);
        int index=0;
        int ncon=((struct vertex_t*) VECTOR(*vertices)[0]) ->ncon;

        for(int i=0;i<vn;++i) {
                for(int j=0;j<ncon;++j) {
                        struct vertex_t *pattr = igraph_vector_ptr_e(vertices,i);
                        /* struct vertex_t *pattr = VECTOR(*vertices)[i]; */
                        igraph_vector_int_push_back(vwgt, pattr->attr[j]);
                }
        }

        for(int i=0;i<vn;++i) {
                igraph_vector_int_push_back(xadj, index);
                igraph_vector_ptr_t *pa = &(al->al[i]);
                int size = igraph_vector_ptr_size(pa);
                for(int j=0;j<size;++j) {
                        struct edge_t *pe = igraph_vector_ptr_e(pa,j);
                        /* struct edge_t *pe = VECTOR(*pa)[j]; */
                        igraph_vector_int_push_back(adjncy, pe->dst);
                        igraph_vector_int_push_back(adjwgt, pe->attr);
                        index++;
                }
        }
        igraph_vector_int_push_back(xadj, index);
#ifndef NDEBUG
        igraph_vector_int_fprint(xadj, stderr);
        igraph_vector_int_fprint(adjncy, stderr);
        igraph_vector_int_fprint(vwgt, stderr);
        igraph_vector_int_fprint(adjwgt, stderr);
#endif 
        return 0;
}

/* Convert from vertices and edges to metis_adjlist
 * Input: vn: number of vertices
 *        en: number of edges
 *      ncon: number of attrs of a single vertex 
 * Ouput: xadj: vertex vid 's adjlist is stored in range 
 *              [xadj[vid], xadj[vid+1]]
 *              len -> vn+1
 *      adjncy: stores adjlists for all vertices continuously
 *      vwgt:   vertex vid 's weight(s) is stored within
 *              [vid * ncon, (vid+1) * ncon]
 *      adjwgt: the edge eid is stored at adjncy[eid]
 *              whose weight is stored at adjwgt[eid]
 * Time: O(vn * en)
 */

int vector2adjlist(int vn, int en, int ncon,
                igraph_vector_ptr_t *vertices,
                igraph_vector_ptr_t *edges,
                idx_t **xadj,
                idx_t **adjncy,
                idx_t **vwgt,
                idx_t **adjwgt)
{
        struct edge_t *pedge;
        int *pattr, dst, src, attr;
        int index=0, subsrp=0;

        *xadj = xmalloc(sizeof(idx_t)*(vn+1));
        *adjncy = xmalloc(sizeof(idx_t)*2*en);
        *vwgt = xmalloc(sizeof(idx_t)*ncon *vn);
        *adjwgt = xmalloc(sizeof(idx_t)*en*2);

        for(int i=0;i<vn;++i) {
                for(int j=0;j<ncon;++j) {
                        int pos = i * ncon  + j ;
                        /* struct vertex_t *pattr = VECTOR(*vertices)[i]; */
                        struct vertex_t *pattr = igraph_vector_ptr_e(vertices,i);
                        (*vwgt)[pos] = pattr->attr[j];
                }
        }
        for(int i=0;i<vn;i++) {
                (*xadj)[subsrp++] = index;
                for(int j=0;j<en;++j) {
                        pedge = VECTOR(*edges)[j];
                        src = pedge->src;
                        dst= pedge->dst;
                        attr = pedge->attr;
                        if(src != i && dst != i) 
                                continue;
                        int vv = src == i ? dst : src;
                        (*adjncy)[index] = vv;
                        (*adjwgt)[index] = attr;
                        ++index;
                        /* printf("%d ", vv); */
                }
                /* puts("\n"); */
        }
        (*xadj)[subsrp] = index;

        return 0;
}

int vector2adjlistEX(igraph_vector_ptr_t *vertices,
                igraph_vector_ptr_t *edges,
                igraph_vector_int_t *xadj,
                igraph_vector_int_t *adjncy,
                igraph_vector_int_t *vwgt,
                igraph_vector_int_t *adjwgt,
                int ncon)
{

        struct edge_t *pedge;
        int *pattr, dst, src, attr;
        int index=0, subsrp=0;
        int vn, en;

        vn =  igraph_vector_ptr_size(vertices);
        en = igraph_vector_ptr_size(edges) ;
        igraph_vector_int_init(xadj, 0);
        igraph_vector_int_init(adjncy, 0);
        igraph_vector_int_init(vwgt, 0);
        igraph_vector_int_init(adjwgt, 0);

        for(int i=0;i<vn;++i) {
                for(int j=0;j<ncon;++j) {
                        int pos = i * ncon  + j ;
                        /* pattr = VECTOR(*vertices)[i]; */
                        struct vertex_t *pattr = igraph_vector_ptr_e(vertices,i);
                        igraph_vector_int_push_back(vwgt, pattr->attr[j]);
                        /* (*vwgt)[pos] = pattr[j]; */
                }
        }

        for(int i=0;i<vn;i++) {
                igraph_vector_int_push_back(xadj, index);
                for(int j=0;j<en;++j) {
                        /* pedge = VECTOR(*edges)[j]; */
                        pedge=igraph_vector_ptr_e(edges,i);
                        src = pedge->src;
                        dst= pedge->dst;
                        attr = pedge->attr;
                        if(src != i && dst != i) 
                                continue;
                        int vv = src == i ? dst : src;
                        igraph_vector_int_push_back(adjncy, vv);
                        igraph_vector_int_push_back(adjwgt, attr);
                        /* VECTOR(*adjncy)[index] = vv; */
                        ++index;
                }
        }
        igraph_vector_int_push_back(xadj, index);

#ifndef NDEBUG
        igraph_vector_int_fprint(xadj, stderr);
        igraph_vector_int_fprint(adjncy, stderr);
        igraph_vector_int_fprint(vwgt, stderr);
        igraph_vector_int_fprint(adjwgt, stderr);

#endif
        return 0;
}

/* print textual info for an adjlist in METIS format */      
void adjlist_printf(int vn, int en, int ncon,
                idx_t *xadj, idx_t *adjncy,
                idx_t *vwgt, idx_t *adjwgt)
{
        printf("xadj:\n");
        for(int i=0;i<vn+1;++i) {
                printf("%d ", xadj[i]);
        }
        puts("\n");
        printf("adjncy:\n");
        for(int i=0;i<2*en;++i) {
                printf("%d ", adjncy[i]);
        }

        puts("\n");


        printf("vwgt:\n");
        for(int i=0;i<vn;++i) {
                printf("%d: ", i);
                for(int j=0;j<ncon;++j) {
                        printf("%d ", vwgt[j]);
                }
                puts("\n");
        }

        puts("\n");

        printf("adjwgt:\n");
        for(int i=0;i<2*en ;++i) {
                printf("%d ", adjwgt[i]);
        }

        puts("\n");
}


void print_partition(FILE *stream, idx_t *part, idx_t nvtxs)
{
        for(int i=0;i<nvtxs;i++) {
                fprintf(stream, "%d %d\n",i+1,  part[i]);
        }
}

int call_vector2adjlist(
                int ncon,
                igraph_vector_ptr_t *vertices,
                igraph_vector_ptr_t *edges
                )
{
        int vn =  igraph_vector_ptr_size(vertices);
        int en = igraph_vector_ptr_size(edges) ;

        idx_t *xadj, *adjncy, *vwgt, *adjwgt;
        int result;
        result = vector2adjlist(vn, en ,
                        ncon,
                        vertices, 
                        edges,
                        &( xadj ),
                        &( adjncy ),
                        &( vwgt ),
                        &( adjwgt )
                        );
        adjlist_printf(vn,en,ncon, xadj, adjncy, vwgt, adjwgt);
        return 0;
}

int dimacs2files(FILE *dimacs, int npart,FILE *vfile, FILE *efile, FILE *rfile)
{
        igraph_t undirected_graph;
        igraph_adjlist_t iadjlist;
        zlog_category_t *mylog = zlog_get_category("mylog");
        
        zlog_info(mylog,"igraph_read_graph_dimacs begin");
        int rs = igraph_read_graph_dimacs(&undirected_graph, dimacs, 0,0,0,0,0,0);
        if(rs != IGRAPH_SUCCESS) {
                zlog_error(mylog, "igraph_read_graph_dimacs failed");
                return -1;
        }
        zlog_info(mylog,"igraph_read_graph_dimacs finish");

        igraph_vector_int_t xadj;
        igraph_vector_int_t adjncy;
        int vn = igraph_vcount(&undirected_graph);
        int en = igraph_ecount(&undirected_graph);
        
        zlog_info(mylog, "Graph to Adjlist begin"); 
        rs = igraph_adjlist_init(&undirected_graph,&iadjlist,IGRAPH_ALL);
        if(rs != IGRAPH_SUCCESS) {
                zlog_error(mylog, "igraph_adjlist_init failed");
                return -1;
        }
        zlog_info(mylog, "Graph to Adjlist finish, vn=%d, en=%d",vn,en);

        zlog_info(mylog, "Preparation for partitioning begin");
        igraph_vector_int_init(&xadj, 0);
        igraph_vector_int_init(&adjncy, 0);
        idx_t *part = calloc(sizeof(idx_t), vn);
        idx_t edgecut;
        int index=0; 
        int ncon=1;
        for(int i=0;i<vn;++i) {
                igraph_vector_int_push_back(&xadj, index);
                igraph_vector_int_t *pa = igraph_adjlist_get(&iadjlist,i);
                int size = igraph_vector_int_size(pa);
                for(int j=0;j<size;++j) {
                        int pe = VECTOR(*pa)[j];
                        igraph_vector_int_push_back(&adjncy, pe);
                        index++;
                }
        }
        igraph_vector_int_push_back(&xadj, index);
        zlog_info(mylog, "Preparation for partitioning finish");

        zlog_info(mylog, "METIS_PartGraphKway begin");
        rs = METIS_PartGraphKway(&vn,&ncon,xadj.stor_begin,adjncy.stor_begin,0,0,0,&npart,0,0,0,&edgecut,part);
        if(rs != METIS_OK) {
                zlog_error(mylog, "METIS_PartGraphKway failed");
                return -1;
        }
        zlog_info(mylog, "METIS_PartGraphKway finish");

        zlog_info(mylog, "Generating rfile...");
        print_partition(rfile, part, vn);
        zlog_info(mylog, "Generatiion done, edgecut is %d", edgecut);

        zlog_info(mylog, "Generating vfile...");

        for(int i=0;i<vn;++i) {
                fprintf(vfile, "%d 0\n", i+1);
        }
        zlog_info(mylog,"Generatiion done, use the vfile with ncon=1, vwgt=NULL, vsize=NULL");

        zlog_info(mylog, "Generating efile...");
        igraph_es_t alles;
        igraph_es_all(&alles, IGRAPH_EDGEORDER_ID);
        igraph_eit_t alleit;
        igraph_eit_create(&undirected_graph,alles,&alleit);
        while(!IGRAPH_EIT_END(alleit))
        {
                int eid = IGRAPH_EIT_GET(alleit);
                int from, to;
                igraph_edge(&undirected_graph,eid,&from,&to);
                fprintf(efile,"%d %d 1\n",from+1,to+1);
                IGRAPH_EIT_NEXT(alleit);
        }
        zlog_info(mylog, "Generatiion done, use the efile with adjwgt=NULL");
/* #ifndef NDEBUG */
/*         zlog_info (mylog, "Showing intermedia result.."); */
/*         igraph_adjlist_fprint(&iadjlist,stderr); */
/*         igraph_vector_int_fprint(&xadj,stderr); */
/*         igraph_vector_int_fprint(&adjncy,stderr); */
/*         print_partition(stderr, part, vn); */
/* #endif */
        igraph_eit_destroy(&alleit);
        free(part);
        igraph_vector_int_destroy(&xadj);
        igraph_vector_int_destroy(&adjncy);
        igraph_destroy(&undirected_graph);
        fclose(dimacs);
        fclose(rfile);
        fclose(vfile);
        fclose(efile);
        return 0;
}

int verify_sssp(FILE *dimacs, int source, igraph_matrix_t *result, FILE *dat)
{
        igraph_t undirected_graph;
        zlog_category_t *mylog = zlog_get_category("mylog");
        int vn, rs;

        rs = igraph_read_graph_dimacs(&undirected_graph,
                        dimacs,
                        0,0,0,0,0,IGRAPH_UNDIRECTED);
        if(rs != IGRAPH_SUCCESS) {
                zlog_error(mylog,"read dimacs failed");
                return -1;
        }
        vn = igraph_vcount(&undirected_graph);
        igraph_vs_t from,to;
        igraph_vs_1(&from, source);
        igraph_vs_all(&to);
        igraph_matrix_init(result,1,vn);

        rs = igraph_shortest_paths(&undirected_graph,result,from,to,IGRAPH_ALL);
        
        if(rs != IGRAPH_SUCCESS) {
                zlog_error(mylog,"igraph_shortest_paths failed");
                return -1;
        }
        igraph_matrix_fprint(result,stderr);
        if(NULL != dat) {
                for(int i=0;i<vn;++i) {
                /* int dist = (int) MATRIX(*result,0,i); */
                int dist = (int) igraph_matrix_e(result,0,i);
                fprintf(dat,"%d\t%d\n", i+1, dist);
                }
                fclose(dat);
        }

        // cleanup
        igraph_destroy(&undirected_graph);

        return 0;
}

/* let clearify the stuffs about metis's input format:
 * line 1 will be header line i.e.
 * (n m [fmt] [ncon])
 * remaining lines describe info about a single vertex 
 * per line in a adjlist favour.
 * 
 * line i: s w1 w2 ... w[ncon] v1 e1 v2 e2 ...
 * 
 * show more details:
 * i: implicit vertex i described about in this line 
 * s: vsize
 * w[i]: vwgt
 * v[i]: vertex adj to i
 * e[i]: adjwgt[i] i.e. weight of edge {i, v[i]}
 * ncon: number of constrains on vertex i
 * about fmt:
 * 001: has adjwgt  (or NULL)
 * 010: has vwgt (or NULL)
 * 100: has vsize (or NULL)
 * if no fmt feild is provided, adjwgt , vwgt and vsize 
 * are assume to be all 1
 * note that comment starts from `%'
 */

/* this function requires fmt string from caller and check
 * against the actual content of struct edge_t and vertex_t
 * to see if fmt is correct.
 * i.e.
 * pv->vsize == 0 -> no vsize
 * pv->ncon == 0 -> no vwgt
 * pe->attr == -1 -> no adjwgt(negative weight is currently not
 * supported)
 * 
 * normally for graph generated from vfile and efile format,
 * fmt will be 011 (with vwgt and adjwgt, without vsize)
 */

int my_adjlist2gpmetis(my_adjlist_t *al, int en,
               igraph_vector_ptr_t *vertices, char *fmt,
              FILE *gpmetis)
{
        struct vertex_t *pv = VECTOR(*vertices)[0];
        struct edge_t *pe = VECTOR(*my_adjlist_get(al,0))[0];
        int ncon = pv->ncon;
        int vn = al->vn;
        int hvsize = pv->vsize != 0;
        int hvwgt = pv->ncon != 0;
        int hadjwgt = pe->attr != -1;

        zlog_category_t *mylog = zlog_get_category("mylog");
        if(fmt[2]=='1' && !hadjwgt) {
                zlog_error(mylog, "adjwgt not found");
                return -1;
        }
        if(fmt[1]=='1' && !hvwgt) {
                zlog_error(mylog, "vwgt not found");
                return -1;
        }
        if(fmt[0]=='1' &&!hvsize) {
                zlog_error(mylog, "vsize not found");
                return -1;
        }

        fprintf(gpmetis, "%d %d %s", vn, en, fmt);
        if(hvwgt)
                fprintf(gpmetis, " %d\n", ncon); // ncon will be 1
        else
                fputs("\n", gpmetis);

        for(int i=0;i<vn;++i) {
                pv = VECTOR(*vertices)[i];
                if(hvsize)
                        fprintf(gpmetis,"%d ", pv->vsize);
                for(int j=0;j<ncon;++j) {
                        fprintf(gpmetis, "%d ", pv->attr[j]);
                }

                igraph_vector_ptr_t *pa = my_adjlist_get(al,i);
                int size = igraph_vector_ptr_size(pa);
                for(int k=0;k<size;++k) {
                        pe = VECTOR(*pa)[k];
                        fprintf(gpmetis,"%d ", pe->dst+1);
                        if(hadjwgt)
                                fprintf(gpmetis, "%d ", pe->attr);
                }
                fputs("\n", gpmetis);
        }

        // check with graphchk in the caller 
        return 0;
}

int va_system(char *fmt,...)
{
        char *cmd;
        size_t size = BUFSIZ;
        va_list ap;
        va_start(ap,fmt);
        while(1) {
                cmd = xmalloc(sizeof(char)*size);
                int sz=vsnprintf(cmd, size, fmt, ap);
                if(sz<0)
                        return -1;
                if(size >= sz)         
                        break;

                size<<=1;
                cmd = realloc(cmd, size);
        }
        int rs = system(cmd);
        free(cmd);
        va_end(ap);
        return rs;
}  

/* this function minics PEval of sssp
 * Input:
 *      IN_messages: igraph_vector_ptr_t  of struct my_pair, 
 *      namely, (u, dist[u]) .
 *      
 *      OUT_messages: igraph_vector_ptr_t  of igraph_vector_ptr_t *
 *      the reason not choosing igraph_matrix_t is that, it is in 
 *      fact igraph_vector_t, so individual rows cannot be extracted.
 *      
 *      Partial_result: igraph_vector_ptr_t of len vn and of my_pair
 *      the reason is  to provide hashmap like directed access to
 *      vertex of vid.
 *      
 *      TODO 
 *      Fragment: this will be a struct that provides 
 *          o: can tell whether a vertex isBorderVertexWithOutgoingEdge
 *          o: can tell whether a vertex isInnerVertex
 *          o: can getOutgoingEdge of u
 *          o: can getInnerVertex
 *          o: can get all the Fragment it communicates to based on
 *             a isBorderVertexWithOutgoingEdge u.
 *          
 *          since these operations is best done with a hashmap that
 *          only add and query. hsearch_r is choosed
 */

/* this section discusses the Fragment:
 * 
 * o: since our graph is undirected
 *  isBorderVertexWithOutgoingEdge(u) -> number of fids that u communicates to > 0
 *  this can be a member of struct vertex_t as each vertex has at least 0 fids that 
 *  communicates to.
 * o: isInnerVertex (u) -> fragtable[u] == self.fid
 * o: getInnerVertex() -> an igraph_vector_ptr_t of vertex_t*
 * o: get all the fids it communicates to: visits v->fids if v->nfid > 0
 * o: getOutgoingEdge(u) -> visits the adjlist of u 
 */

#define N_FRAG 4
igraph_vector_int_t fragtable;

void print_fragments(FILE *s, struct Fragment *fragments, int nfrag)
{
        struct vertex_t *pv;
        fprintf(s,"fragments(nfrag=%d\n\n",nfrag);
        fputs("-----------------------------------------------------------------------------------\n", s);
        for(int i=0;i<nfrag;++i) {
                fprintf(s,"\tfragment(fid=%d)\n\n\tlocal_adjlist=\n", fragments[i].fid);
                int size=igraph_vector_ptr_size((fragments[i].innerv));
                fputs("\t\t", s);
                my_adjlist_print(s,fragments[i].adjlist);
                fprintf(s, "\tinner_vertices=\n");
                for(int j=0;j<size;++j) {
                        fputs("\t\t", s);
                        /* struct vertex_t *pv=VECTOR(*(fragments[i].innerv))[j]; */
                        pv=igraph_vector_ptr_e(fragments[i].innerv, j);
                        print_vertex(s,pv);
                }
                fputs("-----------------------------------------------------------------------------------\n", s);
        }
        fputs(")\n", s);
}

/* for convinience, code a iterator struct for visiting 
 * innerv of a fragment
 */

struct vertex_t *getInnerVertex(struct Fragment *fragment)
{

        int index=fragment->innerv_it;
        return fragment->innerv->stor_begin[index];
}

struct vertex_t *nextInnerVertex(struct Fragment *fragment)
{
        int index= fragment->innerv_it++;
        return fragment->innerv->stor_begin[index];
}

int endInnerVertex(struct Fragment *fragment)
{
        return fragment->innerv_it == igraph_vector_ptr_size(fragment->innerv);
}

/* check if vid belongs to fid, that is, isInnerVertex of fid */
/* NO CHECK */

int isInnerVertex(int fid, int vid, igraph_vector_int_t *fragtable)
{
       int belongs = VECTOR(*fragtable)[vid];
        return fid == belongs;
}

/* MESSAGE BUFFER SECTION */


void print_msg_buf(FILE *s, struct msg_buf_t *msg)
{
        int nfrag=msg->nfrag;
        fprintf(s,"msg_buf_t(nfrag=%d)\ninmsg=(inmsg_size=%ld)\n",msg->nfrag,
                        igraph_vector_ptr_size(msg->inmsg));
        int size=igraph_vector_ptr_size(msg->inmsg);
        for(int i=0;i<size;++i) {
                print_pair(s,igraph_vector_ptr_e(msg->inmsg,i));
        }
        fprintf(s,"\noutmsg=(outmsg_size=%ld)\n",
                        igraph_vector_ptr_size(msg->outmsg));
        
        for(int i=0;i<nfrag;++i) {
                fprintf(s,"\nTO fid=%d\n", i);
                void *ptr=igraph_vector_ptr_e(msg->outmsg,i);
                int size=igraph_vector_ptr_size(ptr);
                for(int j=0;j<size;++j) {
                        print_pair(s,igraph_vector_ptr_e(ptr,j));
                }
        }
        fputs("\n", s);
}
        
void print_msg_buf_all(FILE *s, struct msg_buf_t *msg)
{
        int nfrag=msg[0].nfrag;

        fputs("BUFFERS\n---------------------------------------------------------\n", s);
        for(int i=0;i<nfrag;++i) {
                fprintf(s,"MESSAGE BUFFER %d\n", i);
                print_msg_buf(s, &msg[i]);
                fputs("---------------------------------------------------------\n", s);
        }
}

void print_pair(FILE *s, struct pair_t *p)
{
        fprintf(s,"my_pair(vid=%d,dist=%d)\n", 
                        p->vid,
                        p->dist);
}

/* about the memory management of the 
 * actual content of MSG, in the sssp
 * case, it is pair_t.
 * workers use IN msg to create OUT msg after evaluation,
 * then its IN msg will be cleared but and
 * destroyed, the OUT will be transferred to IN
 * by coordinator, then the OUT will be cleared
 * by not destroyed as the IN now keeps ref to the MSG
 */

/* destroy and clear refs */
void clear_in_msg(struct msg_buf_t *msg)
{
        // destroy its all of its elements with free
        igraph_vector_ptr_t *inmsg=msg->inmsg;
        // free_all + destroy == destroy_all
        igraph_vector_ptr_clear(inmsg);
        // make it empty
        
#ifndef NDEBUG
        assert(igraph_vector_ptr_empty(inmsg));
#endif

}

/* clear but not destroy, using igraph_vector_ptr_resize to achieve */
/* never use igraph_vector_ptr_clear OR igraph_vector_ptr_free_all */ 

void clear_all_out_msg(struct msg_buf_t *msg)
{
        // clear outmsg but not destroy its elements
        int nfrag=msg->nfrag;

        for(int i=0;i<nfrag;++i) {
                void *ptr=igraph_vector_ptr_e(msg->outmsg, i);
                /* igraph_vector_ptr_clear(ptr); */
                igraph_vector_ptr_resize(ptr,0);
#ifndef NDEBUG
                assert(igraph_vector_ptr_empty(ptr));
#endif
        }
}

/* forward declaration */
int add_in_msg(struct msg_buf_t*,void*);
igraph_vector_ptr_t *get_out_msg_at(struct msg_buf_t *msg , int index);
int get_out_msg_size_at(struct msg_buf_t *msg, int fid);

int sync_message(struct args_t *args, int nfrag)
{ 
        struct msg_buf_t *msg;
        // clear all IN msg
        for(int i=0;i<nfrag;++i) {
                clear_in_msg(args[i].messages);
        }

        for(int i=0;i<nfrag;++i) {

                /* igraph_vector_ptr_t *each_outmsg=igraph_vector_ptr_e(args[i].messages->outmsg,i); */
                igraph_vector_ptr_t *each_outmsg=get_out_msg_at(args[i].messages, i);
                for(int j=0;j<nfrag;++j) {
                       /* igraph_vector_ptr_t *fid_outmsg=igraph_vector_ptr_e(each_outmsg,j); */ 
                       igraph_vector_ptr_t *fid_outmsg=get_out_msg_at(args[i].messages, j);
                       int size=get_out_msg_size_at(args[i].messages,j);

                       assert(fid_outmsg);
                       /* int size=igraph_vector_ptr_size(fid_outmsg); */
                       for(int k=0;k<size;k++) {
                               void *piece_outmsg=igraph_vector_ptr_e(fid_outmsg,k);
                               /* igraph_vector_ptr_push_back(args[j].messages->inmsg, piece_outmsg); */
                               add_in_msg(args[j].messages, piece_outmsg);
                       }
                }
                clear_all_out_msg(args[i].messages);
        }
        
        return 0;
}

int add_out_msg_fid(struct msg_buf_t *msg, int fid, struct pair_t *pp)
{
        zlog_category_t *mylog = zlog_get_category("mylog");
        /* if(fid < 0 || fid >= msg->nfrag) { */
        /*         zlog_error(mylog, "invalid fid %d", fid); */
        /*         return -1; */
        /* } */
        igraph_vector_ptr_t *ptr=igraph_vector_ptr_e(msg->outmsg, fid);
#ifndef NDEBUG
        assert(ptr);
#endif
        igraph_vector_ptr_push_back(ptr, pp);
        return 0;
}

igraph_vector_ptr_t *get_out_msg_at(struct msg_buf_t *msg , int index)
{
        zlog_category_t *mylog = zlog_get_category("mylog");
        /* if(index < 0 || index >= msg->nfrag) { */
        /*         zlog_error(mylog, "invalid index %d", index); */
        /*         return NULL; */
        /* } */
        return igraph_vector_ptr_e(msg->outmsg, index);
}

int add_in_msg(struct msg_buf_t *msg, void *ptr)
{
#ifndef NDEBUG
        assert(msg->inmsg);
#endif
        return igraph_vector_ptr_push_back(msg->inmsg, ptr);
}

void *get_in_msg_at(struct msg_buf_t *msg, int index)
{

        zlog_category_t *mylog = zlog_get_category("mylog");

        void *rs=igraph_vector_ptr_e(msg->inmsg, index);
        assert(rs);

        return rs;
}

int get_in_msg_size(struct msg_buf_t *msg)
{
        return igraph_vector_ptr_size(msg->inmsg);
}

int get_out_msg_size_at(struct msg_buf_t *msg, int fid)
{
        
        zlog_category_t *mylog = zlog_get_category("mylog");
        /* if(fid < 0 || fid >= msg->nfrag) { */
        /*         zlog_error(mylog, "invalid fid %d", fid); */
        /*         return -1; */
        /* } */
        igraph_vector_ptr_t *ptr=igraph_vector_ptr_e(msg->outmsg, fid);

#ifndef NDEBUG
        assert(ptr);
#endif

        return igraph_vector_ptr_size(ptr);
}

struct edge_t *make_edge(int src, int dst, int attr,int eid)
{
        struct edge_t *pe=xmalloc(sizeof(struct edge_t));
        pe->src=src;
        pe->dst=dst;
        pe->attr=attr;
        pe->eid=eid;
        return pe;
}

/* load_fragment: Creates N_FRAG fragments 
 * and a global adjlist from vertices,
 * edges and fragtable 
 * Time: O(vn+en+N_FRAG)
 */

int load_fragment(
                igraph_vector_ptr_t *vertices,
                igraph_vector_ptr_t *edges,
                igraph_vector_int_t *fragtable,
                my_adjlist_t *adjlist,
                struct Fragment *fragments
                )
{
        int vn=igraph_vector_ptr_size(vertices);
        int en=igraph_vector_ptr_size(edges);
        struct vertex_t *pv;
        struct edge_t *pe;
        int vid,belongs,fid;
        int src, dst;
        igraph_vector_ptr_t* pf;
 
        // init fragments 
        for(int i=0;i<N_FRAG;++i) {
                fragments[i].innerv=xmalloc(sizeof(igraph_vector_ptr_t));
                fragments[i].fid=i;
                igraph_vector_ptr_init( fragments[i].innerv , 0);
                fragments[i].innerv_it=0;
                fragments[i].adjlist=xmalloc(sizeof(my_adjlist_t));
        }

        // visits vertices and build innerv
        for(int i=0;i<vn;++i) {
                pv = VECTOR(*vertices)[i];
                vid=pv->vid;
                belongs=VECTOR(*fragtable)[vid];
                pv->fid=belongs;
                pf=fragments[belongs].innerv;
                pv->locid=igraph_vector_ptr_size(pf);
                igraph_vector_ptr_push_back(pf, pv);
        }
        

        // build global adjlist 
        vector2my_adjlist(vertices,edges,adjlist); 

        // build local adjlist 
        for(int i=0;i<N_FRAG;++i) {
                int innerv_size= igraph_vector_ptr_size(fragments[i].innerv);
                fragments[i].innerv_size=innerv_size;
                my_adjlist_init_empty(fragments[i].adjlist, innerv_size); 
                fragments[i].gadjlist=adjlist;
        }
        
        // for each v, filter out the u that is not in the same fragment with him
        // in his adjlist 
        for(int i=0;i<vn;++i) {
                igraph_vector_ptr_t *global_adjlist=my_adjlist_get(adjlist,i); // global_adjlist to be scaned
                int size=igraph_vector_ptr_size(global_adjlist);
                pv=igraph_vector_ptr_e(vertices,i); 
                int src_fid=pv->fid;
                int local_vid=pv->locid;
                igraph_vector_ptr_t *local_adjlist=my_adjlist_get( fragments[src_fid].adjlist, local_vid);
                // local_adjlist to be filled
                for(int j=0;j<size;++j) {
                        struct edge_t *pe, *local_pe;
                        pe=igraph_vector_ptr_e(global_adjlist,j); // fetch an edge from global_adjlist
                        int dst=pe->dst;
                        struct vertex_t *pv_dst = igraph_vector_ptr_e(vertices, dst);
                        int dst_fid=pv_dst->fid;
                        if(dst_fid != src_fid)
                                continue;
                        // not within the same fragment
                        
                        local_pe=make_edge(local_vid,pv_dst->locid,pe->attr,pe->eid);
                        igraph_vector_ptr_push_back(local_adjlist,local_pe);
                }
        }

        // visits edges and build nfid and fids of vertex_t 
        // use an igraph_matrix_bool_t to mark if a fid 
        // has been add to a vertex's fids list
        
        igraph_matrix_bool_t hasadd;
        igraph_matrix_bool_init(&hasadd,vn,N_FRAG);
        igraph_bool_t *pbool;
        for(int i=0;i<en;++i) {
                /* pe=VECTOR(*edges)[i]; */
                pe=igraph_vector_ptr_e(edges,i);
                src=pe->src;
                dst=pe->dst;
                int srcbelong=VECTOR(*fragtable)[src];
                int dstbelong=VECTOR(*fragtable)[dst];
                if(srcbelong == dstbelong)
                        continue; // they are not border vertices

                // add to v's fids list
                pbool=igraph_matrix_bool_e_ptr(&hasadd,src,dstbelong);
                if(! *pbool) {
                        *pbool = 1;
                        /* pv=VECTOR(*vertices)[src]; */
                        pv=igraph_vector_ptr_e(vertices,src);
                        igraph_vector_int_push_back(pv->fids, dstbelong);
                }

                pbool=igraph_matrix_bool_e_ptr(&hasadd, dst,srcbelong);
                if(! *pbool) {
                        *pbool=1;
                        /* pv=VECTOR(*vertices)[dst]; */
                        pv=igraph_vector_ptr_e(vertices,dst);
                        igraph_vector_int_push_back(pv->fids, srcbelong);
                }
        }
#ifndef NDEBUG
        print_fragments(stderr,fragments, N_FRAG);
#endif 
        igraph_matrix_bool_destroy(&hasadd);
        return 0;
} 

/* before develop sssp_peval, first check whether running dijktra
 * on the whole graph yield the same result as igraph_simple_paths
 */
void sssp_whole(my_adjlist_t *adjlist,int source,igraph_vector_int_t *result)
{
        igraph_indheap_t heap;
        igraph_vector_bool_t visited;
        igraph_vector_ptr_t *al;
        int dist,src,dst,attr;
        struct vertex_t *pv;
        struct edge_t *pe;
        
        // zero based
        source--;

        int vn= adjlist->vn;
        igraph_vector_int_init(result,vn);
        igraph_vector_bool_init(&visited,vn);
        igraph_indheap_init(&heap,vn);

        // deal with source
        for(int i=0;i<vn;++i)
                *igraph_vector_int_e_ptr(result, i)=INT_MAX;
        VECTOR(*result)[source]=0;
        igraph_indheap_push_with_index(&heap,source,0);
        
        // main loop
        while(!igraph_indheap_empty(&heap))
        {
                int u=igraph_indheap_max_index(&heap);
                /* dist=igraph_indheap_max(&heap); */
                dist= - igraph_indheap_delete_max(&heap);

                if(VECTOR(visited)[u])
                        continue;
                VECTOR(visited)[u]=1;

                al=my_adjlist_get(adjlist,u);
                int size=igraph_vector_ptr_size(al);

                for(int i=0;i<size;++i) {
                        /* pe=VECTOR(*al)[i]; */
                        pe=igraph_vector_ptr_e(al,i);
                        dst=pe->dst;
                        attr=pe->attr;
                        /* if(attr + dist < VECTOR(*result)[dst]) */
                        if(attr + dist < igraph_vector_int_e(result,dst))
                        {
                                *igraph_vector_int_e_ptr(result,dst) = attr + dist;
                                /* VECTOR(*result)[dst] = attr + dist; */
                                igraph_indheap_push_with_index(&heap,dst,(igraph_real_t) -(attr + dist));
                        }
                }
        }

#ifndef NDEBUG
        igraph_vector_int_fprint(result,stderr);
#endif
        igraph_indheap_destroy(&heap);
        igraph_vector_bool_destroy(&visited);
}

struct pair_t *my_make_pair(int vid, int dist)
{
        struct pair_t *pp=xmalloc(sizeof(struct pair_t));
        pp->dist=dist;
        pp->vid=vid;
        return pp;
}

/* sssp_peval:
 * do sssp partial evaluation on a fragment 
 * Input: RESULT, FRAGMENT, MSG, GADJLIST
 * Output: NONE
 */

void sssp_peval(void *ptr)
{

        // get actuall argumets from void*
        struct args_t *args=ptr;
        struct Fragment *fragment= args->fragment;
        igraph_vector_int_t *result=args->result;
        /* igraph_vector_int_t *fragtable=args->fragtable; */
        struct msg_buf_t *msg=args->messages;
        struct vertex_t *source=args->source;
        int myfid=fragment->fid;
        my_adjlist_t *adjlist=fragment->adjlist;
        int vn=adjlist->vn;
        igraph_indheap_t heap;
        igraph_vector_bool_t visited;
        int innerv_size=igraph_vector_ptr_size(fragment->innerv);
        
        assert(result); 
        igraph_vector_int_fill(result, INT_MAX);

        if(source->fid != myfid) 
                return;

        igraph_vector_bool_init(&visited, innerv_size);
        /* *igraph_vector_int_e_ptr(result,source_locid)=0; */
        igraph_vector_int_set(result,source->locid,0);
        igraph_indheap_init(&heap,0);
        igraph_indheap_push_with_index(&heap, source->locid, 0);

        // main loop 
        while(!igraph_indheap_empty(&heap))
        {
                int u=igraph_indheap_max_index(&heap);
                int dist= (int)- igraph_indheap_delete_max(&heap);
                if(VECTOR(visited)[u]) 
                        continue;

                VECTOR(visited)[u]=1;

                // issue OUT_messages
                struct vertex_t * pv=igraph_vector_ptr_e(fragment->innerv,u);
                int nfid=igraph_vector_int_size(pv->fids);

                for(int i=0;i<nfid;++i) {
                        int fid=igraph_vector_int_e(pv->fids,i);
                        /* igraph_vector_ptr_t *msg2fid=igraph_vector_ptr_e(outmsg,fid); */
                        struct pair_t *pp=my_make_pair(pv->vid,dist);
                        /* igraph_vector_ptr_push_back(msg2fid, pp); */
                        add_out_msg_fid(msg, fid, pp);
                }
                
                // visit the local adjlist of u
                igraph_vector_ptr_t *local_adjlist=my_adjlist_get(fragment->adjlist,u);
                int adjsize=igraph_vector_ptr_size(local_adjlist);

                for(int i=0;i<adjsize;++i) {
                        struct edge_t *pe=igraph_vector_ptr_e(local_adjlist,i);
                        int dst=pe->dst;
                        int attr=pe->attr;
                        int distv=igraph_vector_int_e(result,dst);
                        if(distv > dist + attr) {
                                int new_dist=dist+attr;
                                igraph_vector_int_set(result, dst, new_dist);
                                igraph_indheap_push_with_index(&heap,dst,(igraph_real_t) -  new_dist);
                        }
                }
        }

}

/* sssp_inceval:
 * updates the partial_result of fragment i
 * based on IN msg designated to i and generates
 * OUT msg to the fragments it communicates to
 * Input: RESULT, FRAGMENT, MSG, GADJLIST
 * Output: NONE
 */

void sssp_inceval(void *ptr)
{

        /* get needed argumets */ 
        struct args_t *args=ptr;
        struct msg_buf_t *msg=args->messages;
        igraph_vector_int_t *result=args->result;
        struct Fragment *fragment=args->fragment;
        int myfid=fragment->fid;
        my_adjlist_t *global_adjlist=fragment->gadjlist;


        struct pair_t *pp;
        igraph_indheap_t heap;
        struct vertex_t *pv;
        struct edge_t *pe;
        int inmsg_size=get_in_msg_size(msg);
        igraph_indheap_init(&heap, 0);

        /* updates partial_result of myfid */
        for(int i=0;i<inmsg_size;++i) {
                pp=get_in_msg_at(msg, i);
                assert(pp);
                igraph_vector_ptr_t *adjlist=my_adjlist_get(global_adjlist, pp->vid);
                int adjlist_size=igraph_vector_ptr_size(adjlist);

                /* visits the global_adjlist of i to find all the end points that falls into myfid */
                for(int j=0;j<adjlist_size;++j) {
                        pe=igraph_vector_ptr_e(adjlist,j);
                        pv=igraph_vector_ptr_e(args->vertices,pe->dst);
                        if(pv->fid != myfid)
                                continue;

                        /* Converts global_vid to local_vid */
                        int local_vid=pv->locid;
                        int dist=igraph_vector_int_e(result,local_vid);
                        if(pe->attr + pp->dist < dist) {
                                int new_dist=pe->attr + pp->dist;
                                igraph_vector_int_set(result, local_vid, new_dist);
                                igraph_indheap_push_with_index(&heap, local_vid, (igraph_real_t) - new_dist);
                        }
                }
        }
        
        igraph_vector_bool_t visited;
        igraph_vector_bool_init(&visited,fragment->innerv_size);

        /* main loop */
        while(!igraph_indheap_empty(&heap))
        {
                int u=igraph_indheap_max_index(&heap);
                int dist= (int) - igraph_indheap_delete_max(&heap);

                if(VECTOR(visited)[u])
                        continue;
                VECTOR(visited)[u]=1;

                // issue OUT_messages
                struct vertex_t * pv=igraph_vector_ptr_e(fragment->innerv,u);
                int nfid=igraph_vector_int_size(pv->fids);
                
                // visits BorderVertexWithOutgoingEdge innerv 
                for(int i=0;i<nfid;++i) {
                        int fid=igraph_vector_int_e(pv->fids,i);
                        /* igraph_vector_ptr_t *msg2fid=igraph_vector_ptr_e(outmsg,fid); */

                        /* Converts to global_vid */
                        pp=my_make_pair(pv->vid,dist);
                        /* igraph_vector_ptr_push_back(msg2fid, pp); */
                        add_out_msg_fid(msg, fid, pp);
                }
                
                // visit the local adjlist of u
                igraph_vector_ptr_t *iadjlist=my_adjlist_get(fragment->adjlist,u);
                int adjsize=igraph_vector_ptr_size(iadjlist);

                for(int i=0;i<adjsize;++i) {
                        pe=igraph_vector_ptr_e(iadjlist,i);
                        int dst=pe->dst;
                        int attr=pe->attr;
                        int distv=igraph_vector_int_e(result,dst);
                        if(distv > dist + attr) {
                                int new_dist=dist+attr;
                                igraph_vector_int_set(result,dst,new_dist);
                                igraph_indheap_push_with_index(&heap,dst,(igraph_real_t) -new_dist);
                        }
                }
        }
        igraph_vector_bool_destroy(&visited);
        igraph_indheap_destroy(&heap);
}


extern FILE *sssp_dat;

/* sssp_assemble:
 * assemble all the partial_results and writes it to file
 * Input : argumets array
 * Ouput: assembled_result in the format of
 * vn line, each line -> vid    dist
 * Time: O(vn)
 */

void sssp_assemble(void *ptr)
{
        struct args_t *args=ptr;
        int vn=args[0].vn;
        int nfrag=args[0].nfrag;
        igraph_vector_int_t  assembled_result;
        igraph_vector_int_t *partial_result;
        FILE *output_file=sssp_dat==NULL?stdout:sssp_dat;

        igraph_vector_int_init(&assembled_result,vn);
        for(int i=0;i<nfrag;++i) {
                partial_result=args[i].result ;
                int size=igraph_vector_int_size(partial_result);
                struct Fragment *fragment=args[i].fragment;
                for(int j=0;j<size;++j) {
                        int local_vid=j;
                        struct vertex_t *pv=igraph_vector_ptr_e(fragment->innerv,j);
                        int global_vid=pv->vid;
                        int dist=igraph_vector_int_e(partial_result,j);
                        /* igraph_vector_int_set(&assembled_result, global_vid,dist); */
                        VECTOR(assembled_result)[global_vid]=dist;
                }
        }
        
        
        for(int i=0;i<vn;++i) {
                int dist=VECTOR(assembled_result)[i];
                fprintf(output_file, "%d\t%d\n", i+1, dist);
        }
        igraph_vector_int_destroy(&assembled_result);


}


int rfile2fragtable(FILE *rfile, int vn, igraph_vector_int_t *fragtable)
{
        int fid,vid;
        igraph_vector_int_init(fragtable,vn);
        while(!feof(rfile))
        {
                fscanf(rfile, "%d %d", &vid, &fid);
                VECTOR(*fragtable)[vid-1]=fid;
        }
        return vn;
}

/* these global variables are associated with run_sssp_on_fragments */


#define SSSP_MAX_STEP 100
/* the args_t array used to pass argumets to peval and inceval */ 
struct args_t sssp_args[N_FRAG];

/* the global adjlist that records the input graph */
my_adjlist_t sssp_global_adjlist;

/* the fragment array that representes the local graph */ 
struct Fragment sssp_fragments[N_FRAG];

/* two parts of the message buffer : IN and OUT */
struct msg_buf_t  sssp_message[N_FRAG];

/* the partial result array that keep partial_result of each fragment during each step */
igraph_vector_int_t sssp_result[N_FRAG];

/* global unique fragment table mapping global vid to its fid */ 
igraph_vector_int_t sssp_fragtable;

/* global vertices and edges arrays that keep track of all the data */
igraph_vector_ptr_t sssp_vertices, sssp_edges;

/* input of graph comes here */ 
FILE *sssp_vfile,*sssp_efile,*sssp_rfile, *sssp_dat;

/* parameters passed to run_on_fragments */ 
struct params_t sssp_params;


/* sync_message: 
 * Input: argumets array and its len nfrag
 * Ouput:
 * IN msg is first cleared, then
 * the OUT msg coming from all fragments
 * to fragment fid will be sent to the IN msg
 * of fid. OUT msg is cleared after that 
 */

/* is_finish:
 * Check if all the work is done, signaled by no new IN msg
 * is generated by sync_message OR reaching max_step
 * Input: argumets array and its len, transfering IN msg
 * Ouput: bool, whether the work is done
 */

int is_finish(struct args_t *args, int nfrag, int super_step, int max_step)
{
        int halt_cnt=0;
        
        for(int i=0;i<nfrag;++i) {
                int size=get_in_msg_size(args[i].messages);
                if(0 == size) 
                        halt_cnt++;
        }
        return (nfrag == halt_cnt || super_step > max_step);
}

/* run_on_fragments:
 * run the algorithm specified by peval, inceval and assemble
 * Input: 
 *      peval: PARTIAL EVALUATION
 *      inceval: INCREMENTAL EIVALUATION
 *      args: argumets array that the algorithm 
 *      will be run on each fragment 
 *      params: other parameters that provided input
 *      or tuning to the function
 */



void print_msg_buf_all(FILE *, struct msg_buf_t*);

int run_on_fragments(   void (*peval)(void*), 
                        void (*inceval)(void*),
                        void (*assemble)(void*),
                        struct args_t *args,
                        struct params_t *params)

{
        
        zlog_category_t *mylog = zlog_get_category("mylog");

        /* vfile2vertices */
        int vn=vfile2vertices(params->vfile,1,params->vertices);
        
        /* efile2edges */
        int en=efile2edges(params->efile, params->edges); 
        
        /* rfile2fragtable */        
        rfile2fragtable(params->rfile,vn, params->fragtable);
        
        /* load_fragment */
        load_fragment(  params->vertices, 
                        params->edges, 
                        params->fragtable, 
                        params->gadjlist, 
                        params->fragments);
        
        struct vertex_t *pv_source = 
                igraph_vector_ptr_e(params->vertices,  params->source -1 );
        
        for(int i=0;i<params->nfrag;++i) {
                args[i].vn=vn;
                args[i].en=en;
                args[i].source=pv_source;
                args[i].data=NULL;
                igraph_vector_int_init(args[i].result, args[i].fragment->innerv_size);
        }  

        int super_step=0;
        igraph_bool_t active=1;
        while(1)
        {
                for(int i=0;i<args->nfrag;++i) {
                        if(super_step==0) { peval(&args[i]); }
                        else { inceval(&args[i]); }
                }
                super_step++;
                print_msg_buf_all(stderr,sssp_message);
                sync_message(args, params->nfrag);
                if(is_finish(args, args->nfrag, super_step, params->max_step))
                        break;

        }
        assemble(args);
        
        return 0;
}

int msg_buf_init(struct msg_buf_t *msg_buf, int nfrag)
{
        msg_buf->nfrag=nfrag;
        msg_buf->outmsg=xmalloc(sizeof(igraph_vector_ptr_t));
        msg_buf->inmsg=xmalloc(sizeof(igraph_vector_ptr_t));
        igraph_vector_ptr_init(msg_buf->outmsg, 0);
        igraph_vector_ptr_init(msg_buf->inmsg,0);
        igraph_vector_ptr_set_item_destructor(msg_buf->inmsg,free);


        for(int j=0;j<nfrag;++j) {
                 
                igraph_vector_ptr_t *ptr= xmalloc(sizeof(igraph_vector_ptr_t));
                igraph_vector_ptr_init(ptr,0);
                igraph_vector_ptr_set_item_destructor(ptr, free);
                /* igraph_vector_ptr_set(msg_buf->outmsg,j,ptr); */
                igraph_vector_ptr_push_back(msg_buf->outmsg, ptr);
        }
        return 0;
} 

void msg_buf_destory(struct msg_buf_t *msg)
{
        void *out=msg->outmsg;
        void *in=msg->inmsg;

        igraph_vector_ptr_destroy_all(in);
        for(int i=0;i<msg->nfrag;++i) {
                igraph_vector_ptr_t *ptr=igraph_vector_ptr_e(out,i);
                igraph_vector_ptr_destroy_all(ptr);
        }
}

int sssp_init()
{
        for(int i=0;i<N_FRAG;++i) {
                sssp_args[i].adjlist=&sssp_global_adjlist;
                sssp_args[i].source=NULL;
                sssp_args[i].result=&sssp_result[i];
                sssp_args[i].fragment=&sssp_fragments[i];
                sssp_args[i].fragtable=&sssp_fragtable;
                msg_buf_init(&sssp_message[i], N_FRAG);
                sssp_args[i].messages=&sssp_message[i];

                sssp_args[i].nfrag=N_FRAG;
                sssp_args[i].vertices=&sssp_vertices;
                sssp_args[i].edges=&sssp_edges;
                sssp_args[i].source=NULL;
        } 
        sssp_params.fragments=sssp_fragments;
        sssp_params.gadjlist=&sssp_global_adjlist;
        sssp_params.nfrag=N_FRAG;
        sssp_params.vertices=&sssp_vertices;
        sssp_params.edges=&sssp_edges;
        sssp_params.fragtable=&sssp_fragtable;
        sssp_params.vfile=fopen("test-dimacs.v", "r");
        sssp_params.efile=fopen("test-dimacs.e", "r");
        sssp_params.rfile=fopen("test-dimacs.r", "r");
        sssp_params.source=1;
        sssp_params.max_step=4;
        return 0;
}

void fragment_free(struct Fragment *fragment)
{
        igraph_vector_ptr_destroy(fragment->innerv);
        my_adjlist_destory(fragment->adjlist);
}

void sssp_cleanup(int nfrag)
{
        /* my_adjlist_destory(&sssp_global_adjlist); */
        /* /1* igraph_vector_ptr_destroy_all(&sssp_vertices); *1/ */

        /* for(int i=0;i<nfrag;++i) { */
        /*         fragment_free(&sssp_fragments[i]); */
        /*         igraph_vector_int_destroy(&sssp_result[i]); */
        /*         msg_buf_destory(&sssp_message[i]); */
        /* } */
}

#if 0

        char *vfilen=input_files[0];
        char *efilen=input_files[1];
        char *rfilen=input_files[2];
        
        FILE * rfile = fopen(vfilen, "r");
        FILE * vfile = fopen(efilen, "r");
        FILE * efile = fopen(rfilen, "r");
#endif

